package com.geek.framework.study.event.zookeeper;

import com.geek.framework.study.event.distributed.MessagingEngineContext;
import com.geek.framework.study.event.io.DefaultDeserializer;
import com.geek.framework.study.event.io.DefaultSerializer;
import com.geek.framework.study.event.io.Deserializer;
import com.geek.framework.study.event.io.Serializer;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.recipes.queue.QueueSerializer;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.io.IOException;

/**
 * @Description zookeeper 消息引擎上下文信息
 */
public class ZookeeperMessagingEngineContext implements MessagingEngineContext {

    private static final QueueSerializer<Object> DEFAULT_QUEUE_SERIALIZER = new QueueSerializer<Object>() {
        private final Serializer serializer = new DefaultSerializer();
        private final Deserializer deserializer = new DefaultDeserializer();

        @Override
        public byte[] serialize(Object item) {
            try {
                return serializer.serialize(item);
            } catch (IOException e) {
                e.printStackTrace();
                return null;
            }
        }

        @Override
        public Object deserialize(byte[] bytes) {
            try {
                return deserializer.deserialize(bytes);
            } catch (IOException e) {
                e.printStackTrace();
                return null;
            }
        }
    };

    private final QueueSerializer<Object> queueSerializer;

    private String connectString = "127.0.0.1:2181";

    private RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);

    /**
     * 消息队列路径
     */
    private String queuePath;

    public ZookeeperMessagingEngineContext(String queuePath) {
        if (!queuePath.startsWith("/")) {
            queuePath = "/" + queuePath;
        }
        this.queuePath = queuePath;
        this.queueSerializer = DEFAULT_QUEUE_SERIALIZER;
    }

    public String getConnectString() {
        return connectString;
    }

    public void setConnectString(String connectString) {
        this.connectString = connectString;
    }

    public RetryPolicy getRetryPolicy() {
        return retryPolicy;
    }

    public void setRetryPolicy(RetryPolicy retryPolicy) {
        this.retryPolicy = retryPolicy;
    }

    public String getQueuePath() {
        return queuePath;
    }

    public void setQueuePath(String queuePath) {
        this.queuePath = queuePath;
    }

    public QueueSerializer<Object> getQueueSerializer() {
        return queueSerializer;
    }
}
